package cn.cas.mango.util.RequestUtil;

import cn.cas.mango.dto.HotListDetail;
import cn.cas.mango.dto.param.HotListDetailParam;
import cn.cas.mango.service.ConfigService;
import cn.cas.mango.service.PartitionIndexService;
import cn.cas.mango.util.TimeUtil;
import cn.cas.mango.util.string.StringUtils;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch.core.CountRequest;
import co.elastic.clients.elasticsearch.core.CountResponse;
import co.elastic.clients.elasticsearch.core.SearchRequest;

import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.json.JsonData;
import lombok.extern.slf4j.Slf4j;
import org.apache.logging.log4j.util.Strings;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Component
@Slf4j
public class HotListDetailRequest {
  @Autowired
  ElasticsearchClient elasticsearchClient;
  @Autowired
  PartitionIndexService partitionIndexService;
  @Autowired
  ConfigService configService;
  private  BoolQuery.Builder wrapBool(HotListDetailParam hotListDetailParam) {
    String startDate = hotListDetailParam.getStartDate();
    String endDate = hotListDetailParam.getEndDate();
    BoolQuery.Builder bool = new BoolQuery.Builder();
    bool.filter(f -> f.range(r -> r.field("insert_time").gte(JsonData.of(startDate))
        .lte(JsonData.of(endDate))));
    if (!hotListDetailParam.getPlatformCode().equals("all")) {
      bool.must(m->m.terms(t->t.field("platform").terms(
          v->v.value(
              hotListDetailParam.getPlatformList().stream().map(x->FieldValue.of(x)).collect(Collectors.toList()))
      )));
    }
    if (!hotListDetailParam.getSubject().equals("all")) {
      bool.must(m->m.terms(t->t.field("subject").terms(
          v->v.value(
              hotListDetailParam.getSubjectList().stream().map(x->FieldValue.of(x)).collect(Collectors.toList())
          )
      )));
    }

    return bool;
  }

  public SearchRequest.Builder buildRequest(HotListDetailParam hotListDetailParam, int from, int size) {
    SearchRequest.Builder builder = new SearchRequest.Builder();
    BoolQuery.Builder bool = wrapBool(hotListDetailParam);
    String startDate = StringUtils.transTime2Date(hotListDetailParam.getStartDate());
    String endDate = StringUtils.transTime2Date(hotListDetailParam.getEndDate());
    String index = configService.getIndex();
    index = partitionIndexService.partitionIndex(index, startDate, endDate);
    builder.index(index).query(q->q.bool(bool.build())).from(from).size(size);
    return builder;
  }

  public long getCount(HotListDetailParam hotListDetailParam) {
    CountRequest.Builder countSearch = new CountRequest.Builder().index(configService.getAliasIndex());
    BoolQuery.Builder bool = wrapBool(hotListDetailParam);
    CountRequest.Builder query = countSearch.query(q -> q.bool(bool.build()));
    TimeUtil timeUtil = new TimeUtil();
    try {
      CountRequest build = query.build();
      log.info("【uuid: {}】 get [startDate: {} endDate: {}] count request: {}", hotListDetailParam.getUuid(),
          hotListDetailParam.getStartDate(), hotListDetailParam.getEndDate(),
          build);
      CountResponse r = elasticsearchClient.count(build);
      long count = r.count();
      log.info("【uuid: {}】 get [startDate: {} endDate: {}] count number: {}, cost: {}s", hotListDetailParam.getUuid(),
          hotListDetailParam.getStartDate(), hotListDetailParam.getEndDate(),
          count, timeUtil.getCost());
      return count;
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

  public List<Hit<HotListDetail>> getNews(HotListDetailParam hotListDetailParam) {
    List<Hit<HotListDetail>> ret = new ArrayList<>();
    int count = (int)getCount(hotListDetailParam);
    count = Math.min(count, hotListDetailParam.getMaxReadingSize());
    log.info("【uuid: {}】, actual reading size: {}", hotListDetailParam.getUuid(), count);
    hotListDetailParam.setTotalNews(count);

    int batchSize = configService.getReadBatchSize();
    int epoch = (int)(Math.ceil(count * 1.0 / batchSize));
    TimeUtil timeUtil = new TimeUtil();
    log.info("【uuid: {}】 reading from es [startDate: {} endDate: {}] start...", hotListDetailParam.getUuid(),
        hotListDetailParam.getStartDate(), hotListDetailParam.getEndDate()
        );
    for (int i = 0; i < epoch; i++) {
      SearchRequest.Builder builder = buildRequest(hotListDetailParam, i * batchSize, batchSize);
      SearchRequest build = builder.build();
      if (i == 0) {
        log.info("【uuid: {}】 reading from es [startDate: {} endDate: {}] search request template: {}", hotListDetailParam.getUuid(),
            hotListDetailParam.getStartDate(), hotListDetailParam.getEndDate(), build
        );
      }
      try {
        SearchResponse<HotListDetail> r = elasticsearchClient.search(build, HotListDetail.class);
        List<Hit<HotListDetail>> list = r.hits().hits();
        ret.addAll(list);
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }
    log.info("【uuid: {}】 reading from es [startDate: {} endDate: {}] reading from es end, total cost: {}s", hotListDetailParam.getUuid(),
        hotListDetailParam.getStartDate(), hotListDetailParam.getEndDate(), timeUtil.getCost()
    );
    if (configService.isClusterFilterNotChinese()) {
      ret = filterNotChinese(ret);
    }
//    ret = filterDuplicate(ret);
    return ret;
  }


  private List<Hit<HotListDetail>> filterDuplicate(List<Hit<HotListDetail>> list) {
    List<Hit<HotListDetail>> ret = new ArrayList<>();
    Map<String, Hit<HotListDetail>> map = new HashMap<>();

    for (Hit<HotListDetail> hit: list) {
      HotListDetail source = hit.source();
      map.put(source.getKey(), hit);
    }
    return map.values().stream().collect(Collectors.toList());
  }


  private List<Hit<HotListDetail>> filterNotChinese(List<Hit<HotListDetail>> list) {
    List<Hit<HotListDetail>> ret = new ArrayList<>();
    for (Hit<HotListDetail> hit: list) {
      HotListDetail source = hit.source();
      if (Strings.isBlank(source.getTitle())) {
        ret.add(hit);
      } else {
        if (StringUtils.judgeChinese(source.getTitle(), configService.getChineseRatio())) {
          ret.add(hit);
        }
      }
    }
  return ret;
  }
}
